package com.baishancloud.log.foglog

import com.baishancloud.log.common.entity._
import com.baishancloud.log.common.util.{ExternalDataFetchUtil, IpInfoUtil, TimeUtil}
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import java.util.Objects
import scala.collection.mutable

/**
 *
 * @author ziqiang.wang 
 * @date 2022/2/15 14:51
 */
class LogParse extends BroadcastProcessFunction[(String, Origin), ExternalDataEntity, Log] {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[LogParse])
  private var subTaskIndex: Int = -1

  // 国家信息表
  private var countryInfo: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()
  // 省份信息表
  private var provinceInfo: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()
  // 运营商信息表
  private var ispInfo: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()
  // node节点信息
  private var nodeInfo: mutable.Map[String, NodeEntity] = mutable.Map[String, NodeEntity]()
  // 服务类型信息
  private var serviceTypeInfo: mutable.Map[String, Byte] = mutable.Map[String, Byte]()
  // 域名信息表
  private var domainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
  // 主机对应svg信息表
  private var svgInfo: mutable.Map[String, String] = mutable.Map[String, String]()


  override def open(parameters: Configuration): Unit = {
    subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    ExternalDataFetchUtil.countryInfoFetchFromFile(countryInfo)
    LOGGER.info(s"countryInfo${subTaskIndex}初始化完成，countryInfo表size为：${countryInfo.size}")
    ExternalDataFetchUtil.provinceInfoFetchFromFile(provinceInfo)
    LOGGER.info(s"provinceInfo${subTaskIndex}初始化完成，provinceInfo表size为：${provinceInfo.size}")
    ExternalDataFetchUtil.ispInfoFetchFromFile(ispInfo)
    LOGGER.info(s"ispInfo${subTaskIndex}初始化完成，ispInfo表size为：${ispInfo.size}")
    ExternalDataFetchUtil.nodeInfoFetchFromFile(nodeInfo)
    LOGGER.info(s"nodeInfo${subTaskIndex}初始化完成，nodeInfo表size为：${nodeInfo.size}")
    ExternalDataFetchUtil.serviceTypeInfoFetchFromFile(serviceTypeInfo)
    LOGGER.info(s"serviceTypeInfo${subTaskIndex}初始化完成，serviceTypeInfo表size为：${serviceTypeInfo.size}")
    ExternalDataFetchUtil.domainInfoFetchFromFile(domainInfo)
    LOGGER.info(s"domainInfo${subTaskIndex}初始化完成，domainInfo表size为：${domainInfo.size}")
    ExternalDataFetchUtil.svgInfoFetchFromFile(svgInfo)
    LOGGER.info(s"svgInfo${subTaskIndex}初始化完成，svgInfo表size为：${svgInfo.size}")
  }

  override def processElement(value: (String, Origin), ctx: BroadcastProcessFunction[(String, Origin), ExternalDataEntity, Log]#ReadOnlyContext, out: Collector[Log]): Unit = {
    try {
      val origin: Origin = value._2
      val complexField0 = origin.complexField0().split("@")
      val complexField1 = origin.complexField1().split("@")

      val domainStr: String = origin.domain()
      val domain: Int = domainInfo.getOrElse(domainStr, DomainEntity.DEFAULT).domainId
      val logTime: Long = origin.requestStartTime().toEpochSecond
      val timeAt5min: String = TimeUtil.timeAtString(logTime, 300)
      val timeAt1hour: String = TimeUtil.timeAtString(logTime, 3600)
      val timeAt1day: String = TimeUtil.longToString(logTime).substring(0, 10)

      val httpCode: Short = origin.httpCode().toShort

      var httpCodeCauseType: Short = 0
      var info302: String = ""
      try {
        info302 = complexField1(12)
      } catch {
        case _: Exception => info302 = ""
      }
      if (httpCode == 302 && info302.startsWith("psquid:")) {
        httpCodeCauseType = 1
      } else if (httpCode == 302 && info302.startsWith("pluban:")) {
        httpCodeCauseType = 2
      }


      val requestRange: String = origin.requestRange()
      val headRequest: Boolean = httpCode == 200 || requestRange.startsWith("bytes=0-")
      val noHeadRequest: Boolean = !headRequest
      var playRequest: Boolean = false
      var noPlayRequest: Boolean = false
      val isToutiao: Boolean = domainStr.endsWith(".douyinvod.com") || domainStr.endsWith(".bytetos.com") || domainStr.endsWith(".bytegecko.com")
      if (isToutiao) {
        val xReqType: String = origin.ext().split("@")(24)
        if (Objects.equals(xReqType, "play")) {
          playRequest = true
          noPlayRequest = false
        } else {
          noPlayRequest = true
          playRequest = false
        }
      } else {
        noPlayRequest = true
        playRequest = false
      }
      val httpRequestType: Byte =
        if (noHeadRequest && noPlayRequest) {
          0
        } else if (headRequest && noPlayRequest) {
          1
        } else if (noHeadRequest && playRequest) {
          2
        } else if (headRequest && playRequest) {
          3
        } else {
          0
        }

      val ipInfoClient: IpInfoEntity = IpInfoUtil.core(origin.ip())
      val country: Int = countryInfo.getOrElse(ipInfoClient.country, CountryEntity.DEFAULT).id
      val province: Int = provinceInfo.getOrElse(ipInfoClient.province, ProvinceEntity.DEFAULT).id
      val isp: Int = ispInfo.getOrElse(ipInfoClient.isp, IspEntity.DEFAULT).id

      val hostname: String = origin.host()
      val node: String = nodeInfo.getOrElse(hostname, NodeEntity.DEFAULT).nodeCode

      val hitStr = origin.hit()
      val cacheStatus: Short = if (hitStr.contains("MISS") || hitStr.startsWith("TCP_PARENT_")) 0 else if (hitStr.contains("PART")) 2 else if (hitStr.contains("HIT")) 1 else -1
      val cacheLocation: Short = if (hitStr.contains("MEM")) 0 else 1

      var serviceUnit: String = ""
      val cacheGroupCode: String = nodeInfo.getOrElse(hostname, NodeEntity.DEFAULT).cacheGroupCode
      if (cacheGroupCode != null && cacheGroupCode.nonEmpty) {
        serviceUnit = cacheGroupCode
      } else {
        serviceUnit = hostname
      }

      val ipInfoServer: IpInfoEntity = IpInfoUtil.core(origin.serverIp())
      val serviceUnitIsp: Int = ispInfo.getOrElse(ipInfoServer.isp, IspEntity.DEFAULT).id
      val serviceUnitProvince: Int = provinceInfo.getOrElse(ipInfoServer.province, ProvinceEntity.DEFAULT).id

      var serviceGroup: String = hostname
      if (svgInfo.contains(hostname)) {
        serviceGroup = svgInfo(hostname)
      } else if (nodeInfo.contains(hostname)) {
        serviceGroup = nodeInfo(hostname).cacheGroupCode
      }

      val serviceType: Short = serviceTypeInfo.getOrElse(hostname, 0.toByte).toShort
      val logType: Short = value._1 match {
        case "nginx" => 1
        case "foglog" => 2
        case "blb" => 4
        case _ => -1
      }
      var sslProtocol: String = ""
      try {
        sslProtocol = complexField0(10)
      } catch {
        case _: Exception =>
      }

      val isParent: Byte = if (origin.isParentLog) 1 else 0

      val isIpv6: Byte = if (origin.ip().contains(":")) 1 else 0

      val isHttps: Byte = if (Objects.equals(origin.scheme().toLowerCase(), "https")) 1 else 0

      val isBigRequest: Byte = if (origin.bodyBytesSent() > 512 * 1024) 1 else 0

      val sumRequestNum: Long = 1L

      val bytesSend: Long = origin.bytesSent()
      val sumTransBytes: Long = bytesSend

      val sumTransTimeMS: Long = origin.responseTime()


      var sumNotRedirectBytes: Long = 0L
      var sumNotRedirectNum: Long = 0L
      if (isToutiao && origin.url().contains("&cdn_type=")) {
        sumNotRedirectBytes = bytesSend
        sumNotRedirectNum = 1L
      }

      val sumFirstPacketTimeMS: Long = 0L

      var sumFirstFrameNum: Long = 0L
      var sumFirstFrameTimeMS: Long = 0L
      try {
        val firstFrameTimeMS: Long = complexField0(19).toLong
        val byteSent: Long = bytesSend
        if ((httpCode == 200 || requestRange.startsWith("bytes=0-")) && byteSent > 128 * 1024 && firstFrameTimeMS > 0) {
          sumFirstFrameNum = 1L
          sumFirstFrameTimeMS = firstFrameTimeMS
        }
      } catch {
        case _: Exception =>
      }

      var sumSslHandshakeTimeMS: Long = 0L
      var sumSslSessionReusedNum: Long = 0L
      try {
        sumSslHandshakeTimeMS = (complexField0(14).toFloat * 1000).toLong
        if (complexField0(15).toByte == 1) sumSslSessionReusedNum = 1L
      } catch {
        case _: Exception =>
      }

      val sumTransSpeedKBps: Long = (if (origin.responseTime() == 0) bytesSend * 1000.0 / 1024 / 1 else bytesSend * 1000.0 / 1024 / origin.responseTime()).formatted("%.0f").toLong

      val sumSlow256KBpsNum: Long = if (sumTransSpeedKBps < 256) 1L else 0L
      val sumSlow512KBpsNum: Long = if (sumTransSpeedKBps < 512) 1L else 0L

      val sumElapsedWithin1SNum: Long = if (origin.responseTime() <= 1000) 1L else 0L
      val sumElapsedWithin10SNum: Long = if (origin.responseTime() <= 10000) 1L else 0L

      val log = Log(timeAt5min, timeAt1hour, timeAt1day, domain, httpCode, httpCodeCauseType, httpRequestType, country, province, isp, node, cacheStatus, cacheLocation, serviceUnit, serviceUnitIsp, serviceUnitProvince, serviceGroup, serviceType, logType, sslProtocol, isParent, isIpv6, isHttps, isBigRequest, sumRequestNum, sumTransBytes, sumTransTimeMS, sumNotRedirectBytes, sumNotRedirectNum, sumFirstPacketTimeMS, sumFirstFrameNum, sumFirstFrameTimeMS, sumSslHandshakeTimeMS, sumSslSessionReusedNum, sumTransSpeedKBps, sumSlow256KBpsNum, sumSlow512KBpsNum, sumElapsedWithin1SNum, sumElapsedWithin10SNum)
      out.collect(log)
    } catch {
      case _: Exception =>
    }
  }

  override def processBroadcastElement(value: ExternalDataEntity, ctx: BroadcastProcessFunction[(String, Origin), ExternalDataEntity, Log]#Context, out: Collector[Log]): Unit = {
    if (value.countryInfo.nonEmpty) {
      countryInfo = value.countryInfo
      LOGGER.info(s"countryInfo${subTaskIndex}接收到广播数据，countryInfo表size为：${countryInfo.size}")
    }
    if (value.provinceInfo.nonEmpty) {
      provinceInfo = value.provinceInfo
      LOGGER.info(s"provinceInfo${subTaskIndex}接收到广播数据，provinceInfo表size为：${provinceInfo.size}")
    }
    if (value.ispInfo.nonEmpty) {
      ispInfo = value.ispInfo
      LOGGER.info(s"ispInfo${subTaskIndex}接收到广播数据，ispInfo表size为：${ispInfo.size}")
    }
    if (value.nodeInfo.nonEmpty) {
      nodeInfo = value.nodeInfo
      LOGGER.info(s"nodeInfo${subTaskIndex}接收到广播数据，nodeInfo表size为：${nodeInfo.size}")
    }
    if (value.serviceTypeInfo.nonEmpty) {
      serviceTypeInfo = value.serviceTypeInfo
      LOGGER.info(s"serviceTypeInfo${subTaskIndex}接收到广播数据，serviceTypeInfo表size为：${serviceTypeInfo.size}")
    }
    if (value.domainInfo.nonEmpty) {
      domainInfo = value.domainInfo
      LOGGER.info(s"domainInfo${subTaskIndex}接受到广播数据，domainInfo表size为：${domainInfo.size}")
    }
    if (value.svgInfo.nonEmpty) {
      svgInfo = value.svgInfo
      LOGGER.info(s"svgInfo${subTaskIndex}接受到广播数据，svgInfo表size为：${svgInfo.size}")
    }
  }
}
